package ex.tableapi;

import org.apache.flink.table.api.Table;

import static org.apache.flink.table.api.Expressions.$;

/**
 * https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/dev/table/tableapi/#groupby-window-aggregation
 * <p>
 * 更换jar ：  https://mvnrepository.com/artifact/org.apache.flink/flink-connector-jdbc/3.1.1-1.17
 */
public class ExInsertInto extends ApiFrame {

    private String outName = "printOutTable";

    public static void main(String[] args) {
        ExInsertInto demo = new ExInsertInto();
        demo.getEnv();
        Table orders = demo.registerMysql("myorder", "orders");
        orders.printSchema();
        System.out.println("===================");

        Table result = demo.exInsertInto(orders);


        demo.registerConsole(demo.createPrintOutDDL(result.getResolvedSchema().toString()), demo.outName);
        result.executeInsert(demo.outName);
    }


    private Table exInsertInto(Table orders) {
        //与flinkTableName对应（ Table orders = demo.registerMysql("myorder", "orders");），此处填写的是内存表，同时会插入到myorder数据库中。  :
//        TableResult tableResult = orders.insertInto("orders").execute();
//        tableResult.print();
        Table result = orders.select($("*"));
        return result;
    }



}
